package cn.kennylee.codehub.mq.lip.integration.rocketmq.listener;

import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.NumberUtil;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONUtil;
import cn.kennylee.codehub.mq.lip.integration.base.AbstractMqListener;
import cn.kennylee.codehub.mq.lip.integration.base.MessageVo;
import cn.kennylee.codehub.mq.lip.integration.base.MqChannel;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.MDC;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.env.Environment;
import org.springframework.lang.NonNull;

import java.util.Set;

/**
 * <p> RocketMq消费者抽象类 </p>
 * <p>Created on 2025/4/16.</p>
 *
 * @author kennylee
 * @since 0.0.1
 */
@Slf4j
@Getter
@Setter
public abstract class AbstractRocketMqListener extends AbstractMqListener<MessageExt>
    implements RocketMQListener<MessageExt>, ApplicationContextAware {

    private static final String GROUP_SUFFIX = "-GROUP";

    private Environment environment;

    /**
     * 设置同1时间最大消费线程数
     *
     * @see RocketMQMessageListener#consumeThreadMax()
     */
    private int consumeThreadMax = 64;

    /**
     * 每个线程消费消息数
     * <p>注：如果consumeThreadMax和consumeThreadNumber都是1就可以单队列，不担心并发问题了</p>
     *
     * @see RocketMQMessageListener#consumeThreadNumber()
     */
    private int consumeThreadNumber = 20;

    /**
     * 消费线程号占位符，多个线程消费时使用
     * 优先级高于 {@link AbstractRocketMqListener#consumeThreadNumber}
     * <p>例如：${rocketmq.consumer.thread-number-holder}</p>
     */
    private String consumeThreadNumberHolder = "";

    /**
     * 消息主题
     */
    private String topic;

    /**
     * 消息标签，主题下的二级分类；可多个，最终会产出selectorExpression
     */
    private Set<String> tags;

    /**
     * 消息选择表达式；优先级比 {@link AbstractRocketMqListener#tags} 高
     */
    private String selectorExpression = "";

    public AbstractRocketMqListener(String consumerGroup, String topic, Set<String> tags) {
        this.topic = topic;
        this.tags = tags;
        // 实测RocketMQ下一个消费者一个消费组，同Topic下不能重复，否则消费会有异常
        setConsumerGroup(consumerGroup);
    }

    public AbstractRocketMqListener(String consumerGroup, String topic, String tag) {
        this(consumerGroup, topic, Set.of(tag));
    }

    public AbstractRocketMqListener(String topic, String tag) {
        this("", topic, tag);
        this.setConsumerGroup(this.generateConsumerGroup());
    }

    public AbstractRocketMqListener(String topic, String[] tags) {
        this("", topic, Set.of(tags));
        this.setConsumerGroup(this.generateConsumerGroup());
    }

    /**
     * <p>获取消费组名</p>
     *
     * @return 消费组名
     */
    protected String generateConsumerGroup() {
        String consumerGroup = this.getClass().getSimpleName() + GROUP_SUFFIX;
        return StrUtil.replace(consumerGroup, "$", "_");
    }

    /**
     * 消息消费处理
     *
     * @param message 消息体
     */
    @Override
    public void onMessage(MessageExt message) {
        boolean success = super.processMessage(message);

        // 处理成功，直接返回
        if (success) {
            return;
        }

        // 处理失败，重试次数小于最大重试次数，记录日志并重试

        // 重试次数大于最大重试次数，记录失败日志
        if (message.getReconsumeTimes() == getMaxRetryTimes()) {
            this.handleMaxRetryExceeded(message);
        } else {
            log.warn("消息处理失败，msgId: {}, 重试次数: {}", getMsgId(), message.getReconsumeTimes());
            // 重试
            retryMessage(message);
        }
    }

    @Override
    protected void setupTraceId(@NonNull MessageExt message) {
        // 从消息properties获取Trace信息
        String traceId = message.getUserProperty(MqChannel.TRACE_ID_KEY);
        // 放入MDC
        if (StrUtil.isBlank(traceId)) {
            traceId = IdUtil.fastSimpleUUID();
            log.debug("设置 Trace ID: {}", traceId);
        } else {
            log.debug("消息已经存在 Trace ID: {}", traceId);
        }
        MDC.put(MqChannel.TRACE_ID_KEY, traceId);
    }

    /**
     * 消息处理失败，重试次数超过最大重试次数
     *
     * @param message 消息体
     */
    @Override
    protected void handleMaxRetryExceeded(MessageExt message) {
        log.error("消息处理失败，重试次数超过最大重试次数，msgId: {}, 重试次数: {}", getMsgId(), message.getReconsumeTimes());
        // TODO 可选择不抛出异常，手动加入死信，记录失败日志，持久化失败的消息等操作
        // 默认抛出异常
        this.retryMessage(message);
    }

    @Override
    protected void retryMessage(MessageExt message) {
        // TODO 重构使用延迟队列，效果更佳，不用阻塞推送新消息
        super.retryMessage(message);
    }

    /**
     * 获取消息ID
     *
     * @return 消息ID
     */
    @Override
    public String getMsgId() {
        return super.getMessage().getMsgId();
    }

    @Override
    public String getChannel() {
        return super.getMessage().getTopic();
    }

    @Override
    public String getRouteKey() {
        return super.getMessage().getTags();
    }

    @Override
    public boolean isManualAck() {
        // RocketMq不支持手动ACK
        throw new UnsupportedOperationException("RocketMq不支持手动ACK");
    }

    /**
     * 构建RocketMQMessageListener
     *
     * @return RocketMQMessageListener
     */
    public RocketMQMessageListener buildRocketMqMessageListener() {
        String resolvedTopic = resolvePlaceholders(getTopic());
        int finalConsumeThreadNumber = getConsumeThreadNumber();
        if (StrUtil.isNotBlank(getConsumeThreadNumberHolder())) {
            finalConsumeThreadNumber = NumberUtil.parseInt(resolvePlaceholders(getConsumeThreadNumberHolder()), finalConsumeThreadNumber);
        }
        return RocketMqMessageListenerImplBuilder.of(resolvedTopic, resolvePlaceholders(getSelectorExpression()), getConsumerGroup())
            .consumeThreadMax(getConsumeThreadMax())
            .consumeThreadNumber(finalConsumeThreadNumber)
            .build();
    }

    protected String resolvePlaceholders(String placeholder) {
        if (environment == null) {
            throw new IllegalStateException("Environment is not set. Please set the application context first.");
        }
        return environment.resolvePlaceholders(placeholder);
    }

    @Override
    public void setApplicationContext(@NonNull ApplicationContext applicationContext) throws BeansException {
        this.environment = applicationContext.getEnvironment();
    }

    /**
     * 消息转换方法
     */
    @Override
    protected MessageVo convertMessageVo(MessageExt messageExt) {
        try {
            String messageContent = new String(messageExt.getBody());
            if (!JSONUtil.isTypeJSON(messageContent)) {
                log.warn("MQ消息内容不是JSON格式，msgId: {}, body: {}", getMsgId(), messageContent);
                return null;
            }
            return convertMessageVo(messageContent);
        } catch (Exception e) {
            log.error("消息转换失败，msgId: {}", messageExt.getMsgId(), e);
            log.error("原消息体内容：{}", new String(messageExt.getBody()));
            return null;
        }
    }
}
